package com.flow.asyncflow.core.core;


import com.alibaba.fastjson.JSON;
import com.flow.asyncflow.core.annotation.FlowInfo;
import com.flow.asyncflow.core.entity.config.AsyncFlowConfigInfo;
import com.flow.asyncflow.core.entity.config.AsyncFlowInfo;
import com.flow.asyncflow.core.entity.core.*;
import com.flow.asyncflow.core.entity.sql.ExecuteSql;
import com.flow.asyncflow.core.enums.ExecuteEnum;
import com.flow.asyncflow.core.node.FlowNodeTask;
import com.flow.asyncflow.core.node.IFlowNode;
import com.flow.asyncflow.core.node.IFlowPubNode;
import com.flow.asyncflow.core.node.PubGetData;
import com.flow.asyncflow.core.utils.configuration.AsyncFlowConfiguration;
import com.flow.asyncflow.core.utils.log.asyncflowlog.AsyncFlowCommon_Log;
import com.flow.asyncflow.core.utils.log.businesslog.AsyncFlowBusinessLog;
import com.flow.asyncflow.core.utils.msg.AsyncFlowSendMsg;
import com.flow.asyncflow.core.utils.save.BaseSave;
import com.flow.asyncflow.core.utils.threadpool.ThreadPoolExecutorMain;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.StopWatch;

import java.lang.annotation.Annotation;
import java.lang.reflect.ParameterizedType;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;


@SuppressWarnings("unchecked")
public abstract class AsyncFlowManager<TRes, TAsyncContext> {

    private final LinkedList<AsyncFlowNodeModel> asyncFlowNodeList = new LinkedList<>();

    /**
     * 流节点集合【公共节点】--可以提供给多个流使用
     */
    private final LinkedHashMap<ExecuteEnum, LinkedHashMap<IFlowPubNode, Boolean>> flowPubNodes = new LinkedHashMap<>();
    /**
     * 流节点集合【普通节点】--各个流内部定义
     */
    private final LinkedHashMap<ExecuteEnum, LinkedList<AsyncFlowNodeModel>> allNodes = new LinkedHashMap<>();

    /**
     * 流节点集合【普通节点】--各个流内部定义
     */
    private final LinkedHashMap<ExecuteEnum, LinkedHashMap<IFlowNode<TAsyncContext>, Boolean>> flowNodes = new LinkedHashMap<>();

    /**
     * 异步流流信息
     */
    private AsyncFlowInfo asyncFlowInfo = new AsyncFlowInfo();

    private static AsyncFlowConfigInfo asyncFlowStaticInfo = null;
    /**
     * 节点流线程池
     */
    private static ExecutorService pool = null;


    //region 初始化

    /**
     * 初始化
     */
    protected AsyncFlowManager() {
        //加载线程池
        if (pool == null) {
            pool = ThreadPoolExecutorMain.getCustomThreadPoolExecutor();

        }
        //加载站点配置信息（数据库、环境、报存地址）
        if (asyncFlowStaticInfo == null) {
            AsyncFlowConfiguration autoConfiguration = new AsyncFlowConfiguration();
            Map<String, String> dataBaseInfo = autoConfiguration.getDataBaseInfo();
            asyncFlowStaticInfo = new AsyncFlowConfigInfo();
            for (Map.Entry<String, String> entry : dataBaseInfo.entrySet()) {
                asyncFlowStaticInfo.getDataBaseInfo().put(entry.getKey(), entry.getValue());
            }
            asyncFlowStaticInfo.setEvn(autoConfiguration.getEvn());
            asyncFlowStaticInfo.setMongoDBConfigStr(autoConfiguration.getMongoDBConfigStr());
            asyncFlowStaticInfo.setMongoDBName(autoConfiguration.getMongoDBName());
            asyncFlowStaticInfo.setMongoDBTableName(autoConfiguration.getMongoDBTableName());
            asyncFlowStaticInfo.setNodeArmUrl(autoConfiguration.getNodeArmUrl());
            asyncFlowStaticInfo.setBusinessArmUrl(autoConfiguration.getBusinessArmUrl());
            asyncFlowStaticInfo.setBusinessLogUrl(autoConfiguration.getBusinessLogUrl());
        }
        //加载当前流信息，名称，负责人等
        if (StringUtils.isBlank(asyncFlowInfo.getFlowName())
                && StringUtils.isBlank(asyncFlowInfo.getFlowMaintainName())
                && asyncFlowInfo.getTimeOutsLimit() == 0) {
            Annotation[] flowInfo = this.getClass().getAnnotations();
            for (Annotation annotation : flowInfo) {
                //获取方法名
                if (annotation instanceof FlowInfo) {
                    asyncFlowInfo.setFlowName(((FlowInfo) annotation).flowName());
                    asyncFlowInfo.setFlowMaintainName(((FlowInfo) annotation).flowMaintainName());
                    asyncFlowInfo.setTimeOutsLimit(((FlowInfo) annotation).timeOutsLimit());
                    asyncFlowInfo.setIgnoreFlowLog(((FlowInfo) annotation).ignoreFlowLog());
                    break;
                }
            }
            for (Map.Entry<String, String> entry : asyncFlowStaticInfo.getDataBaseInfo().entrySet()) {
                asyncFlowInfo.getDataBaseInfo().put(entry.getKey(), entry.getValue());
            }
            asyncFlowInfo.setEvn(asyncFlowStaticInfo.getEvn());
            asyncFlowInfo.setMongoDBConfigStr(asyncFlowStaticInfo.getMongoDBConfigStr());
            asyncFlowInfo.setMongoDBName(asyncFlowStaticInfo.getMongoDBName());
            asyncFlowInfo.setMongoDBTableName(asyncFlowStaticInfo.getMongoDBTableName());
            asyncFlowInfo.setNodeArmUrl(asyncFlowStaticInfo.getNodeArmUrl());
            asyncFlowInfo.setBusinessArmUrl(asyncFlowStaticInfo.getBusinessArmUrl());
            asyncFlowInfo.setBusinessLogUrl(asyncFlowStaticInfo.getBusinessLogUrl());
        }

    }
    //endregion


    //region 添加节点到执行流中

    /**
     * 添加普通节点
     *
     * @param executeEnum 操作类型
     * @param flowNode    带具体类型的操作节点
     * @param isCanAsync  是否异步
     */
    protected void addFlowNode(ExecuteEnum executeEnum, IFlowNode<TAsyncContext> flowNode, boolean isCanAsync) {

        LinkedHashMap<IFlowNode<TAsyncContext>, Boolean> nodes = flowNodes.computeIfAbsent(executeEnum, k -> new LinkedHashMap<>());
        if (flowNode == null || nodes.containsKey(flowNode)) {
            return;
        }
        if (executeEnum == ExecuteEnum.Async) {
            isCanAsync = true;
        }
        nodes.put(flowNode, isCanAsync);
    }

    /**
     * 添加公共节点
     *
     * @param executeEnum 操作类型
     * @param flowNode    公共操作节点
     * @param isCanAsync  是否异步
     */
    protected void addPubFlowNode(ExecuteEnum executeEnum, IFlowNode flowNode, boolean isCanAsync) {
        LinkedHashMap<IFlowNode<TAsyncContext>, Boolean> nodes = flowNodes.computeIfAbsent(executeEnum, k -> new LinkedHashMap<>());
        if (flowNode == null || nodes.containsKey(flowNode)) {
            return;
        }
        nodes.put(flowNode, isCanAsync);
    }


    //endregion

    //region 执行节点

    //region 执行节点【带耗时明细】

    /**
     * 执行节点【带耗时明细】
     *
     * @param executeEnum      操作类型
     * @param asyncFlowRequest 上下文
     * @param asyncFlowData    请求参数
     * @return AsyncFlowNodeResponse
     */
    private AsyncFlowNodeResponse exeNode(ExecuteEnum executeEnum, AsyncFlowRequest asyncFlowRequest, AsyncFlowContext<TAsyncContext> asyncFlowData, LinkedHashMap<ExecuteEnum, Long> timeConsuming, AsyncFlowResponse<TRes> response, boolean allNodeIsSuccess) {
        AsyncFlowNodeResponse asyncFlowNodeResponse;
        if (executeEnum == ExecuteEnum.Async) {
            pool.execute(() -> {
                exeNode(ExecuteEnum.Async, asyncFlowRequest, asyncFlowData);
                AsyncFlowBusinessLog.saveBusinessLog(asyncFlowRequest, asyncFlowData, response.getMsg(), allNodeIsSuccess);
                //执行顺序 验证-->查询-->操作-->保存-->同步-->异步
                //当执行的操作是异步时，表示当前流所有的操作已经执行完成
                AsyncFlowCommon_Log.SaveExecuteTrackLogs(asyncFlowRequest, asyncFlowData, response, timeConsuming, allNodeIsSuccess, asyncFlowInfo);
            });
            return new AsyncFlowNodeResponse(200, "成功");
        }
        StopWatch sw = new StopWatch();
        sw.start(executeEnum.getMsg() + "-开始");
        //是否走测试流程
        boolean isTest = buildTestData(executeEnum, asyncFlowRequest, asyncFlowData);
        if (isTest) {
            return new AsyncFlowNodeResponse(200, "成功");
        } else {
            asyncFlowNodeResponse = exeNode(executeEnum, asyncFlowRequest, asyncFlowData);
        }

        sw.stop();
        long time = sw.getTotalTimeMillis();
        if (executeEnum == ExecuteEnum.Save && timeConsuming.containsKey(ExecuteEnum.Save)) {
            time = timeConsuming.get(ExecuteEnum.Save) + sw.getTotalTimeMillis();
        }
        timeConsuming.put(executeEnum, time);
        return asyncFlowNodeResponse;
    }

    private boolean buildTestData(ExecuteEnum executeEnum, AsyncFlowRequest asyncFlowRequest, AsyncFlowContext<TAsyncContext> asyncFlowData) {

        boolean isTest = false;
        if (asyncFlowData.getAsyncFlowInfo().isTest() && executeEnum == ExecuteEnum.Sync) {
            return true;
        }
        //测试环境，用线上的数据进行测试，数据源直接用请求里面的json
        if (asyncFlowData.getAsyncFlowInfo().getEvn().equals("test")
                && StringUtils.isNotBlank(asyncFlowData.getTESTContext())
                && StringUtils.isNotBlank(asyncFlowData.getTESTRequest())
                && (executeEnum == ExecuteEnum.Validator || executeEnum == ExecuteEnum.GetData)) {
            if (executeEnum == ExecuteEnum.Validator) {
                return true;
            }
            String data = asyncFlowData.getTESTContext();
            String reqStr = asyncFlowData.getTESTRequest();
            if (StringUtils.isBlank(data) || StringUtils.isBlank(reqStr)) {
                return false;
            }
            Class<TAsyncContext> clazz = (Class<TAsyncContext>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[1];
            asyncFlowData.setContext(JSON.parseObject(data, clazz));
            asyncFlowData.setTESTContext(null);
            asyncFlowData.setTESTRequest(null);
            isTest = true;
        }
        this.asyncFlowInfo.setTest(isTest);
        asyncFlowData.getAsyncFlowInfo().setTest(isTest);
        return isTest;
    }
    //endregion

    //region 执行节点【基础执行】

    /**
     * 执行节点
     *
     * @param executeEnum      操作类型
     * @param asyncData        上下文
     * @param asyncFlowRequest 请求参数
     * @return AsyncFlowNodeResponse
     * @author lgq07612
     **/
    private AsyncFlowNodeResponse exeNode(ExecuteEnum executeEnum, AsyncFlowRequest asyncFlowRequest, AsyncFlowContext<TAsyncContext> asyncData) {
        AsyncFlowNodeResponse asyncFlowNodeResponse = new AsyncFlowNodeResponse(200, "成功");
        List<FlowNodeTask<TAsyncContext>> taskList = new ArrayList<>();
        LinkedHashMap<IFlowNode<TAsyncContext>, Boolean> nodes = flowNodes.get(executeEnum);
        //执行查询时将公共查询一起执行
        if (executeEnum == ExecuteEnum.GetData) {
            LinkedHashMap<IFlowNode<TAsyncContext>, Boolean> pubNodes = flowNodes.get(ExecuteEnum.GetDataPub);
            if (pubNodes != null && !pubNodes.isEmpty()) {
                if (nodes == null) {
                    nodes = new LinkedHashMap<>();
                }
                nodes.putAll(pubNodes);
            }
        }
        if (nodes == null || nodes.isEmpty()) {
            return asyncFlowNodeResponse;
        }
        for (Map.Entry<IFlowNode<TAsyncContext>, Boolean> entry : nodes.entrySet()) {
            //需要异步执行
            if (entry.getValue() || executeEnum == ExecuteEnum.Async) {
                taskList.add(new FlowNodeTask<>(entry.getKey(), asyncFlowRequest, asyncData, entry.getValue(), asyncFlowInfo, executeEnum));
            } else {
                //异步执行集合里面还有未执行完成的任务
                asyncFlowNodeResponse = getTaskListFuture(executeEnum, asyncFlowRequest, taskList);
                if (asyncFlowNodeResponse.getCode() != 200 || asyncFlowNodeResponse.isBreakOff()) {
                    return asyncFlowNodeResponse;
                }
                taskList = new ArrayList<>();
                asyncFlowNodeResponse = new FlowNodeTask<>(entry.getKey(), asyncFlowRequest, asyncData, entry.getValue(), asyncFlowInfo, executeEnum).flowNodeTaskExecute();
                if (asyncFlowNodeResponse.getCode() != 200 || asyncFlowNodeResponse.isBreakOff()) {
                    return asyncFlowNodeResponse;
                }
            }
        }
        asyncFlowNodeResponse = getTaskListFuture(executeEnum, asyncFlowRequest, taskList);
        //只有在非异步操作是需要跳出（异步操作之间互不影响，异步操作的失败，不影响后面的异步操作）
        if (executeEnum != ExecuteEnum.Async && (asyncFlowNodeResponse.getCode() != 200 || asyncFlowNodeResponse.isBreakOff())) {
            return asyncFlowNodeResponse;
        }
        if (executeEnum == ExecuteEnum.GetData) {
            setStrongTypeData(asyncFlowRequest, asyncData);
        }
        return asyncFlowNodeResponse;
    }

    //endregion

    //region 获取异步集合的结果

    /**
     * 获取异步集合的结果
     *
     * @param executeEnum 执行类型
     * @param taskList    集合
     * @return 节点返回
     */
    private AsyncFlowNodeResponse getTaskListFuture(ExecuteEnum executeEnum, AsyncFlowRequest asyncFlowRequest, List<FlowNodeTask<TAsyncContext>> taskList) {
        AsyncFlowNodeResponse asyncFlowNodeResponse = new AsyncFlowNodeResponse(200, "成功");
        if (taskList == null || taskList.size() == 0) {
            return asyncFlowNodeResponse;
        }
        try {
            List<Future<AsyncFlowNodeResponse>> futureList = pool.invokeAll(taskList,1000, TimeUnit.SECONDS);
            for (Future<AsyncFlowNodeResponse> future : futureList) {
                asyncFlowNodeResponse = future.get();
                //只有在非异步操作是需要跳出（异步操作之间互不影响，异步操作的失败，不影响后面的异步操作）
                if (executeEnum != ExecuteEnum.Async && (asyncFlowNodeResponse.getCode() != 200 || asyncFlowNodeResponse.isBreakOff())) {
                    return asyncFlowNodeResponse;
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            String s = AsyncFlowSendMsg.buildErrorMsg(asyncFlowRequest, e, "获取异步集合的结果" + executeEnum.getMsg(), asyncFlowInfo);
            AsyncFlowSendMsg.sendQYWXMsg(asyncFlowInfo.getFlowMaintainName(), s, asyncFlowInfo.getNodeArmUrl());
        }
        return asyncFlowNodeResponse;
    }
    //endregion

    //endregion

    //region 查询操作强类型赋值

    /**
     * 强类型赋值
     *
     * @param asyncFlowRequest 请求
     * @param asyncFlowData    上下文
     */
    private void setStrongTypeData(AsyncFlowRequest asyncFlowRequest, AsyncFlowContext<TAsyncContext> asyncFlowData) {

        try {
            Date startTime = new Date();
            new PubGetData().SetPubData(null, asyncFlowData.getContext(), "", asyncFlowData.getDataMap());
            if (asyncFlowData.getAsyncFlowInfo().getEvn().equals("stage")) {
                AsyncFlowCommon_Log.AddAsyncFlowDetailLog(asyncFlowData, "PubGetData", "数据库上下文", asyncFlowRequest, asyncFlowData.getContext(), true);
                AsyncFlowCommon_Log.AddAsyncFlowLog(asyncFlowRequest, new AsyncFlowNodeResponse(200, "成功"), asyncFlowData, new PubGetData(), startTime, ExecuteEnum.GetData, false);
            }
        } catch (IllegalAccessException | NoSuchFieldException e) {
            String s = AsyncFlowSendMsg.buildErrorMsg(asyncFlowRequest, e, "上下文强类型赋值", asyncFlowInfo);
            AsyncFlowSendMsg.sendQYWXMsg(asyncFlowInfo.getFlowMaintainName(), s, asyncFlowInfo.getNodeArmUrl());
        }
    }
    //endregion

    //region 注入执行流、设置返回参数【子类重写】

    /**
     * 注入执行计划
     */
    protected abstract void managerStartup();

    /**
     * 设置异步流返回
     *
     * @param asyncData             上下文
     * @param asyncFlowNodeResponse 异步流节点返回
     * @param allNodeIsSuccess      是否全部成功
     * @return 返回
     */
    protected abstract AsyncFlowResponse<TRes> setResponse(AsyncFlowContext<TAsyncContext> asyncData, AsyncFlowNodeResponse asyncFlowNodeResponse, boolean allNodeIsSuccess);

    //endregion

    //region 执行流

//    public   AsyncFlowResponse<TRes> managerExec(Object objectReq, EnumBaseInterface reqModuleEnum,EnumBaseInterface reqNodeEnum) throws IllegalAccessException, InstantiationException {
//        AsyncFlowContext<TAsyncContext> orderData = new AsyncFlowContext<>();
//        TAsyncContext context = (TAsyncContext) orderData.getContext().getClass().newInstance();
//        orderData.setContext(context);
//        AsyncFlowRequest request = new AsyncFlowRequest();
//        request.setNodeRequest(objectReq);
//        request.setAsyncFlowModuleEnum(reqModuleEnum);
//        request.setAsyncFlowNodeEnum(reqNodeEnum);
//        return exec(request, orderData);
//    }


    /**
     * 执行异步流
     *
     * @param asyncFlowRequest 请求
     * @param asyncFlowData    上下文
     * @return 返回
     */
    public AsyncFlowResponse<TRes> exec(AsyncFlowRequest asyncFlowRequest, AsyncFlowContext<TAsyncContext> asyncFlowData) {
        AsyncFlowResponse<TRes> response = null;
        AsyncFlowNodeResponse asyncFlowNodeResponse;
        boolean allNodeIsSuccess = false;
        LinkedHashMap<ExecuteEnum, Long> timeConsuming = new LinkedHashMap<>();
        int validateCode = 200, getDataCode = 200, operatorCode = 200, customSaveCode = 200, baseSaveCode = 200, syncCode = 200;
        String nodeErrorMsg = "";
        //是否保存日志
        boolean isSaveOrderLog = false;
        //注入执行计划
        asyncFlowData.setAsyncFlowInfo(asyncFlowInfo);
        asyncFlowData.setDataBaseInfo(asyncFlowStaticInfo.getDataBaseInfo());
        managerStartup();
        //验证
        asyncFlowNodeResponse = exeNode(ExecuteEnum.Validator, asyncFlowRequest, asyncFlowData, timeConsuming, null, false);
        validateCode = asyncFlowNodeResponse.getCode();
        //查询
        if (validateCode == 200 && !asyncFlowNodeResponse.isBreakOff()) {
            asyncFlowNodeResponse = exeNode(ExecuteEnum.GetData, asyncFlowRequest, asyncFlowData, timeConsuming, null, false);
            getDataCode = asyncFlowNodeResponse.getCode();
            //操作
            if (getDataCode == 200) {
                asyncFlowNodeResponse = exeNode(ExecuteEnum.Operator, asyncFlowRequest, asyncFlowData, timeConsuming, null, false);
                operatorCode = asyncFlowNodeResponse.getCode();
                if (operatorCode != 200) {
                    //操作执行错误信息
                    nodeErrorMsg = asyncFlowNodeResponse.getMsg();
                }
                if ((operatorCode == 200 && !asyncFlowNodeResponse.isBreakOff() && asyncFlowData.getExecuteSQLs().size() > 0) || asyncFlowData.getExecuteSQLs().stream().anyMatch(ExecuteSql::isMustExecute)) {
                    asyncFlowNodeResponse = new BaseSave().execute(asyncFlowRequest, asyncFlowData, operatorCode, asyncFlowInfo, timeConsuming);
                    baseSaveCode = asyncFlowNodeResponse.getCode();
                    if (baseSaveCode == 200) {
                        asyncFlowNodeResponse = exeNode(ExecuteEnum.Save, asyncFlowRequest, asyncFlowData, timeConsuming, null, false);
                        customSaveCode = asyncFlowNodeResponse.getCode();
                    }
                    //只要有一个保存成功就需要保存订单日志
                    if (baseSaveCode == 200) {
                        isSaveOrderLog = true;
                    }
                    if (operatorCode == 200 && customSaveCode == 200 && baseSaveCode == 200) {
                        asyncFlowNodeResponse = exeNode(ExecuteEnum.Sync, asyncFlowRequest, asyncFlowData, timeConsuming, null, false);
                        syncCode = asyncFlowNodeResponse.getCode();
                        if (syncCode != 200) {
                            //操作执行错误信息
                            nodeErrorMsg = asyncFlowNodeResponse.getMsg();
                        }
                    }
                }
            }
        }
        if (validateCode == 200 && getDataCode == 200 && operatorCode == 200 && baseSaveCode == 200 && customSaveCode == 200 && syncCode == 200) {
            allNodeIsSuccess = true;
            asyncFlowNodeResponse.setMsg("成功");
        }
        //只有操作的错误信息可能会被保存操作给覆盖掉，因为操作失败了也可能走保存
        if (operatorCode != 200 || syncCode != 200 || baseSaveCode != 200) {
            asyncFlowNodeResponse.setMsg(nodeErrorMsg);
        }
        if (validateCode == -1 || getDataCode == -1 || operatorCode == -1 || syncCode == -1 || baseSaveCode == -1 || customSaveCode == -1) {
            asyncFlowNodeResponse.setMsg("系统异常");
        }
        response = setResponse(asyncFlowData, asyncFlowNodeResponse, allNodeIsSuccess);
        response.setMsg(allNodeIsSuccess ? "成功" : asyncFlowNodeResponse.getMsg());
        response.setCode(asyncFlowNodeResponse.getCode());
        asyncFlowData.setAsyncFlowResponse(response);
        exeNode(ExecuteEnum.Async, asyncFlowRequest, asyncFlowData, timeConsuming, response, allNodeIsSuccess);
        return response;
    }
    //endregion
}
